package com.sunsys.hdfs.compression.gzip;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;

public class DecompressFile {

	public static void main(String []args) {
		
		Configuration configuration = null;
		FileSystem fileSystem = null;
		CompressionCodec compressionCodec = null;
		CompressionCodecFactory compressionCodecFactory = null;
		
		String hdfsUri = "hdfs://localhost:9000/";
		String localSourceFileName = "/home/ubuntu/Desktop/Oct-2012 2.pdf";
		String hdfsCompressionedFileName = "/usr/hdfs/compress/cmpfile1.gz";
		
		FileOutputStream localFileOutputStream = null;
		CompressionInputStream compressionInputStream = null;
		FSDataInputStream fsDataInputStream = null;
		try {
			configuration = new Configuration();
			fileSystem = FileSystem.get(new URI(hdfsUri), configuration);
			compressionCodecFactory = new CompressionCodecFactory(configuration);
			
			Path compressionedFilePath = new Path(hdfsCompressionedFileName);
			fsDataInputStream = fileSystem.open(compressionedFilePath);
			
			
			/* get CompressionCodec instance associated with the file extension */
			compressionCodec = compressionCodecFactory.getCodec(compressionedFilePath);
			compressionInputStream = compressionCodec.createInputStream(fsDataInputStream);
			
			localFileOutputStream = new FileOutputStream(localSourceFileName);
			IOUtils.copyBytes(compressionInputStream, localFileOutputStream, configuration, false);
			
			IOUtils.closeStream(compressionInputStream);
			IOUtils.closeStream(localFileOutputStream);
			IOUtils.closeStream(fsDataInputStream);
		}catch(URISyntaxException e1) {
			e1.printStackTrace();
		}catch(IOException e2) {
			e2.printStackTrace();
		}finally {
		}
		
		System.out.println("Thank you");
		
	}
	
}
